;;; Copyright © 2020 Maxime Devos <maxime.devos@student.kuleuven.be>
;;; This file is part of rehash.
;;;
;;; rehash is free software; you can redistribute it and/or modify it
;;; under the terms of the GNU General Public License as published by
;;; the Free Software Foundation; either version 3 of the License, or (at
;;; your option) any later version.
;;;
;;; rehash is distributed in the hope that it will be useful, but
;;; WITHOUT ANY WARRANTY; without even the implied warranty of
;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
;;; GNU General Public License for more details.
;;;
;;; You should have received a copy of the GNU General Public License
;;; along with rehash.  If not, see <http://www.gnu.org/licenses/>.
(define-library (rehash operations)
  (export rehash? connect-rehash disconnect-rehash
          wait-for-closed-operation wait-for-closed

          query? start-query! abort-query!
          wait-for-query-aborted-operation wait-for-query-aborted
          query-result-operation query-next-result

          ;; don't export a wait-for-store-aborted,
          ;; try wait-for-store-done.
          store? store-mapping! abort-store!
          wait-for-store-aborted-operation
          wait-for-store-done-operation wait-for-store-done)
  (import (rnrs records syntactic)
          (rnrs base)
          (only (srfi srfi-1) delete!)
          (only (rnrs control) when)
          (only (ice-9 optargs) define*)
          (only (fibers) spawn-fiber)
          (only (fibers operations)
                wrap-operation choice-operation perform-operation)
          (only (fibers conditions)
                make-condition wait-operation wait signal-condition!)
          (only (fibers channels)
                make-channel put-operation get-operation)
          (only (rehash)
                query! query-context? query-abort!
                store-hash! store-context? store-abort!
                store-context?
                open-rehash-service close-rehash-service)
          (only (rehash fibers) do-gnunet-stuff))

  (begin
    (define (do-gnunet-stuff/async thunk)
      "Do @var{thunk} in GNUnet context, but don't
wait for it to be scheduled."
      (spawn-fiber
       (lambda ()
         (do-gnunet-stuff thunk))))

    ;; A connection to the rehash service,
    ;; possibly not yet realised.
    (define-record-type (rehash-service make-rehash-service rehash?)
      (fields (immutable close-cvar  rehash-close-cvar)
              (immutable closed-cvar rehash-closed-cvar)
              ;; signalled once lowlevel has been set
              (immutable opened-cvar rehash-opened-cvar)
              ;; only mutated from GNUnet context
              (mutable things-pending rehash-things-pending
                       set-rehash-things-pending!)
              ;; likewise, and #f when closed, '() if not created
              ;; yet.
              (mutable lowlevel rehash-lowlevel set-rehash-lowlevel!))
      (opaque #t)
      (sealed #t))

    (define-record-type (context %make-context context?)
      (fields (immutable abort-cvar context-abort-cvar)
              (immutable aborted-cvar context-aborted-cvar)
              ;; #f when aborted, '() if not created yet.
              ;; Only mutated from GNUnet context
              ;; Set to #f by rehash-loop.
              ;; Set from '() to something else by other 'loop'
              ;; functions.
              (mutable lowlevel context-lowlevel set-context-lowlevel!))
      (opaque #t)
      (sealed #f))

    ;; A representation of a query, possibly not yet started,
    ;; in progress or aborted.
    (define-record-type (query-context make-query-context query?)
      (fields ;; A channel. Don't use from GNUnet context!
       (immutable positive-responses query-positive-responses))
      (opaque #t)
      (sealed #t)
      (parent context))

    ;; A representation of an insertion of a hash in some
    ;; nebulously defined store, possibly not yet started,
    ;; in progress or aborted.
    (define-record-type (store-context make-store-context store?)
      (fields
       ;; Signalled when the insertion has been completed.
       ;; Not signalled when aborted.
       ;; TODO standards for completion
       (immutable done-cvar store-done-cvar))
      (opaque #t)
      (sealed #t)
      (parent context))

    (define (x-lowlevel x?)
      (lambda (q)
        (assert (x? q))
        (context-lowlevel q)))
    (define (x-abort-cvar x?)
      (lambda (q)
        (assert (x? q))
        (context-abort-cvar q)))
    (define (x-aborted-cvar x?)
      (lambda (q)
        (assert (x? q))
        (context-aborted-cvar q)))
    (define (set-x-lowlevel! x? y?)
      (lambda (x y)
        (assert (x? x))
        ;; Perform some validation
        (case (context-lowlevel x)
          ;; stopped -> stopped (identity)
          ((#f) (assert (eq? y #f)))
          ;; unstarted -> started or stopped
          ((()) (assert (or (eq? y #f) (y? y))))
          ;; started -> stopped
          (else (assert (eq? y #f))))
        ;; Seems ok
        (set-context-lowlevel! x y)))

    (define query-lowlevel (x-lowlevel query?))
    (define query-aborted-cvar (x-aborted-cvar query?))
    (define query-abort-cvar (x-abort-cvar query?))
    (define set-query-lowlevel! (set-x-lowlevel! query? query-context?))

    (define store-lowlevel (x-lowlevel store?))
    (define store-abort-cvar (x-abort-cvar store?))
    (define store-aborted-cvar (x-aborted-cvar store?))
    (define set-store-lowlevel! (set-x-lowlevel! store? store-context?))

    ;; Only called from GNUnet context
    ;; Tell fiber-loop the underlying pointer of @var{q}
    ;; should be removed (set to #f) and @var{q} be
    ;; marked aborted if @var{rehash} is closing.
    (define (add-thing-pending! rehash q)
      (assert (rehash? rehash))
      (assert (rehash-has-lowlevel? rehash))
      (assert (context? q))
      (set-rehash-things-pending!
       rehash (cons q (rehash-things-pending rehash))))

    ;; (define-record-type store-context)
    
    (define (connect-rehash configuration)
      "Connect to the rehash,
and return a rehash-service handle.
Forming the actual connection is an
asynchronuous operation.

@var{configuration} configuration for
connecting to the service"
      (let ((s (make-rehash-service (make-condition)
                                    (make-condition)
                                    (make-condition)
                                    '()
                                    '())))
        (spawn-fiber
         (lambda ()
           (rehash-loop s configuration)))
        s))

    (define (remove-thing-pending! rehash q)
      (assert (rehash? rehash))
      (assert (rehash-has-lowlevel? rehash))
      (assert (context? q))
      (set-rehash-lowlevel! rehash
                            (delete! q (rehash-things-pending rehash))))

    (define (disconnect-rehash rehash)
      "Disconnect from the rehash service.
Any active requests will be aborted.
This is an asynchronuous operation.

@var{rehash} handle to the rehash service"
      (signal-condition! (rehash-close-cvar rehash)))

    (define (wait-for-closed-operation rehash)
      "Return an operation for waiting until the
rehash service is closed (or scheduled to be
soonly)."
      (wait-operation (rehash-closed-cvar rehash)))
    (define (wait-for-close-operation rehash)
      (wait-operation (rehash-close-cvar rehash)))

    (define (wait-for-closed rehash)
      (perform-operation (wait-for-closed-operation rehash)))
    (define (wait-for-opened-operation rehash)
      (wait-operation (rehash-opened-cvar rehash)))

    (define (tag-operation tag operation)
      (wrap-operation operation
                      (lambda () tag)))

    (define (x-has-lowlevel? getter)
      (lambda (s)
        (case (getter s)
          ((#f ()) #f)
          (else #t))))
    (define rehash-has-lowlevel?
      (x-has-lowlevel? rehash-lowlevel))
    (define query-has-lowlevel?
      (x-has-lowlevel? query-lowlevel))
    (define store-has-lowlevel?
      (x-has-lowlevel? store-lowlevel))
    
    (define (rehash-loop s config)
      ;; do-gnunet-stuff would work as well here,
      ;; in which case the ‘when’ could be omitted.
      ;; TODO: perhaps do that?
      (do-gnunet-stuff/async
       (lambda ()
         (assert (not (rehash-has-lowlevel? s)))
         (when (eq? '() (rehash-lowlevel s)) ;; could be closed already
           (set-rehash-lowlevel! s (open-rehash-service config))
           (signal-condition! (rehash-opened-cvar s)))))
      (wait (rehash-close-cvar s))
      (do-gnunet-stuff
       (lambda ()
         (when (eq? '() (rehash-lowlevel s)) ;; not even opened yet!
           (set-rehash-lowlevel! s #f))
         (when (rehash-has-lowlevel? s)
           (close-rehash-service (rehash-lowlevel s))
           (set-rehash-lowlevel! s #f)
           (for-each
               (lambda (ctx)
                 (assert (context? ctx))
                 (set-context-lowlevel! ctx #f)
                 (signal-condition! (context-aborted-cvar ctx)))
             (rehash-things-pending s)))
         (signal-condition! (rehash-closed-cvar s)))))

    (define (wait-for-query-abort-operation query)
      (wait-operation (query-abort-cvar query)))

    (define* (start-query! rehash input output-type
                           #:key (anonymity-level #f)
                           (options #f))
      "Asynchronuously start a query for hash->hash mappings.
TODO: no unbounded queues! Place a bound on queue sizes
(maybe implement semaphores), and suspend the search
for hash->hash mappings until the queue is half-empty
or so."
      (let* ((q (make-query-context (make-condition)
                                    (make-condition)
                                    '()
                                    (make-channel))))
        (spawn-fiber
         (lambda ()
           (query-loop q rehash input output-type
                       #:anonymity-level anonymity-level
                       #:options options)))
        q))

    (define* (query-loop q rehash input output-type
                         #:key (anonymity-level #f)
                         (options #f))
      (let ((found-result/gnunet
             (lambda (result)
               (spawn-fiber
                (lambda ()
                  ;; TODO perhaps implement bounded queues
                  ;; and suspend a query if the queue is full
                  ;; (and resume when the queue is short)
                  (perform-operation
                   (choice-operation
                    (put-operation (query-positive-responses q) result)
                    ;; this one is technically superfluous
                    (wait-for-query-abort-operation q)
                    (wait-for-query-aborted-operation q)))))))
            (until-ready-for-creation
             (choice-operation
              (tag-operation 'early-abort
                             (wait-for-query-abort-operation q))
              (tag-operation 'rehash-closed
                             (wait-for-closed-operation rehash))
              ;; technically superfluous, but the earlier
              ;; the better I guess
              (tag-operation 'rehash-close
                             (wait-for-close-operation rehash))
              (tag-operation 'rehash-open
                             (wait-for-opened-operation rehash))))
            (until-death
             (choice-operation
              ;; In this case the lowlevel has already been set to #f
              ;; by fiber-loop
              (tag-operation 'rehash-closed
                             (wait-for-closed-operation rehash))
              ;; In this case the lowlevel should be manually aborted
              (tag-operation 'please-abort-query
                             (wait-for-query-abort-operation q))))
            (properly-abort
             (lambda ()
               ;; fiber-loop could have aborted
               ;; the query already because the rehash
               ;; service is closing
               (when (query-has-lowlevel? q)
                 (remove-thing-pending! rehash q)
                 (query-abort! (query-lowlevel q))
                 ;; Properly mark the query as aborted.
                 (set-query-lowlevel! q #f)
                 (signal-condition! (query-aborted-cvar q))))))
        (case (perform-operation until-ready-for-creation)
          ((early-abort rehash-closed rehash-close)
           (set-query-lowlevel! q #f)
           ;; Aborted before the query even started!
           (signal-condition! (query-aborted-cvar q)))
          ((rehash-open)
           ;; Start the query
           (do-gnunet-stuff
            (lambda ()
              (cond
               ((not (rehash-has-lowlevel? rehash))
                ;; the rehash service was closed
                ;; before the query could be submitted.
                (set-query-lowlevel! q #f)
                (signal-condition! (query-aborted-cvar q)))
               (else
                (set-query-lowlevel! q
                  (query! (rehash-lowlevel rehash)
                          input output-type
                          #:anonymity-level anonymity-level
                          #:options options
                          #:when-found found-result/gnunet))
                ;; ask fiber-loop to remove the pointer
                ;; when the rehash service is closed.
                (add-thing-pending! rehash q)
                ;; Abort the query when asked.
                ;; But don't sleep in GNUnet context.
                (spawn-fiber
                 (lambda ()
                   (case (perform-operation until-death)
                     ((rehash-closed) (values))
                     ((please-abort-query)
                      (do-gnunet-stuff properly-abort)))))))))))))

    (define (wait-for-query-aborted-operation query)
      "An operation for waiting until a query is aborted"
      (assert (query? query))
      (wait-operation (query-aborted-cvar query)))

    (define (wait-for-query-aborted query)
      "Wait until the query is aborted"
      (perform-operation (wait-for-query-aborted-operation query)))

    (define (abort-query! query)
      "Stop a query! The search will be aborted sometime soon.
Due to concurrency issues, it is still possible a result
will still be found (but only soonish if at all)."
      (assert (query? query))
      (signal-condition! (query-abort-cvar query)))

    (define (query-result-operation query)
      "An operation with as return value the next
<query-result>. This operation may never return
if the query is aborted."
      (get-operation (query-positive-responses query)))

    (define (query-next-result query)
      "Wait for a query result and return it.
If the query is aborted before something could
be found, return #f instead"
      (perform-operation
       (choice-operation (query-result-operation query)
                         (tag-operation
                          #f
                          (wait-for-query-aborted-operation query)))))

    (define* (store-mapping! rehash input output
                             #:key
                             (expiration-time #f)
                             (anonymity-level #f)
                             (content-priority #f)
                             (replication-level #f))
      "Store an hash->hash mapping into the data store
 (and DHT, anonymity permitting), to be shared with
 other peers. If the entry is already present, the old
 entry is considered faulty and is replaced with the
 new mapping.

A @code{store?} object is returned that can be used
to abort the insertion.  See the manual for information
on how long the mapping can be expected to stay in
the data store.

The callback @var{when-done} may be called from C and should terminate
quickly.

TODO write the manual
TODO document other pararameters
TODO type-check all parameters early
@param rehash handle to the rehash store
@param input hash to query with
@param output hash to find"
      (assert (rehash? rehash))
      ;; TODO what to do if out of bounds?
      (assert (or (not replication-level)
                  (integer? replication-level)))
      (assert (or (not content-priority)
                  (integer? content-priority)))
      (assert (integer? anonymity-level))
      ;; TODO type-check hashes
      ;; TODO strict types for hashes
      (let ((store (make-store-context (make-condition)
                                       (make-condition)
                                       '()
                                       (make-condition))))
        (spawn-fiber
         (lambda ()
           (store-loop store rehash input output
                       #:expiration-time expiration-time
                       #:anonymity-level anonymity-level
                       #:content-priority content-priority
                       #:replication-level replication-level)))
        store))

    (define (abort-store! store)
      "Stop a store! It will be aborted sometime soon.
Due to concurrency issues, it is still possible the store
will still be completed (but only soonish if at all)."
      (assert (store? store))
      (signal-condition! (store-abort-cvar store)))

    (define (wait-for-store-done-operation store)
      "An operation that waits until a hash->hash mapping
insertion has been completed. Aborts don't count."
      (assert (store? store))
      (wait-operation (store-done-cvar store)))

    (define (wait-for-store-done store)
      "Wait until the hash->hash mapping insertion has been performed,
or the insertion has been aborted. In the former case, return #t,
in the latter case, return #f. The former case is preferred."
      (perform-operation
       (choice-operation
        (tag-operation #t
                       (wait-for-store-done-operation store))
        (tag-operation #f
                       (wait-for-store-aborted-operation store)))))

    (define (wait-for-store-abort-operation store)
      (wait-operation (store-abort-cvar store)))

    (define (wait-for-store-aborted-operation store)
      "An operation for waiting until a hash->hash insertion is aborted.
Completion doesn't count."
      (assert (store? store))
      (wait-operation (store-aborted-cvar store)))

    (define* (store-loop store rehash input output
                         #:key expiration-time
                         anonymity-level
                         content-priority
                         replication-level)
      (let ((until-ready-for-creation
             ;; see comments in query-loop
             (choice-operation
              (tag-operation 'early-abort
                             (wait-for-store-abort-operation store))
              (tag-operation 'rehash-closed
                             (wait-for-closed-operation rehash))
              (tag-operation 'rehash-close
                             (wait-for-close-operation rehash))
              (tag-operation 'rehash-open
                             (wait-for-opened-operation rehash))))
            (until-death
             (choice-operation
              ;; In this case the lowlevel has already been set to #f
              ;; by fiber-loop
              (tag-operation 'rehash-closed
                             (wait-for-closed-operation rehash))
              ;; In this case the lowlevel should be manually aborted
              (tag-operation 'please-abort-store
                             (wait-for-store-abort-operation store))
              ;; nothing to do except cleanup!
              (tag-operation 'completed
                             (wait-for-store-done-operation store))))
            (signal-done
             (lambda ()
               ;; The lowlevel is now useless
               ;; (and the C data structures have been
               ;; freed behind the scenes), so mark
               ;; that properly first.
               (set-store-lowlevel! store #f)
               (remove-thing-pending! rehash store)
               ;; It's completed, not aborted.
               ;; No need to spawn a fiber to
               ;; signal a condition.
               (signal-condition! (store-done-cvar store))))
            (properly-abort
             (lambda ()
               ;; rehash-loop may have
               ;; aborted the store already
               (when (store-has-lowlevel? store)
                 (remove-thing-pending! rehash store)
                 (store-abort! (store-lowlevel store))
                 (set-store-lowlevel! store #f)
                 (signal-condition! (store-aborted-cvar store))))))
        (case (perform-operation until-ready-for-creation)
          ((early-abort rehash-closed rehash-close)
           ;; Store is aborted before it was even started!
           (set-store-lowlevel! store #f)
           (signal-condition! (store-aborted-cvar store)))
          ((rehash-open)
           (do-gnunet-stuff
            (lambda ()
              (cond
               ((not (rehash-has-lowlevel? rehash))
                ;; the rehash service was closed
                ;; before the insertion could be submitted
                (set-store-lowlevel! store #f)
                (signal-condition! (store-aborted-cvar rehash #f)))
               (else
                (set-store-lowlevel! store
                 (store-hash! (rehash-lowlevel rehash)
                              input output
                              #:when-done signal-done
                              #:expiration-time expiration-time
                              #:anonymity-level anonymity-level
                              #:content-priority content-priority
                              #:replication-level replication-level))
                ;; ask fiber-loop to remove the pointer
                ;; when the rehash service is closed.
                (add-thing-pending! rehash store)
                ;; wait for abortion,
                ;; but don't sleep in GNUnet context!
                (spawn-fiber
                 (lambda ()
                   (case (perform-operation until-death)
                     ((rehash-closed completed) (values))
                     ((please-abort-store)
                      (do-gnunet-stuff properly-abort)))))))))))))))

;;;; Local Variables:
;;;; eval: (put 'rehash-lowlevel 'scheme-indent-function 1)
;;;; eval: (put 'for-each 'scheme-indent-function 1)
;;;; eval: (put 'set-query-lowlevel! 'scheme-indent-function 1)
;;;; eval: (put 'set-store-lowlevel! 'scheme-indent-function 1)
;;;; End:
